package mc.support.zmq.route;

import mc.support.proto.ZmqRegisterReq;
import mc.support.proto.ZmqRegisterRsp;
import mc.support.task.common.CronTaskAdaptor;
import mc.support.task.common.Waitable;
import mc.support.util.PlainUtil;
import mc.support.zmq.ProtoConstants;
import mc.support.zmq.ProtoHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZMQ;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import static mc.support.proto.ZmqRegisterReq.Ops;

/**
 * Created with IntelliJ IDEA. <br/>
 * Author: Francis Yuen    <br/>
 * Date: 2014-10-27  <br/>
 */
public class ZmqServerRegisterClient extends CronTaskAdaptor implements ZmqServerRegister, ProtoConstants {

    private static final Logger log = LoggerFactory.getLogger( ZmqServerRegisterClient.class );

    private final Set<String> clients;
    private final Map<String, List<String>> registered = new HashMap<>();
    private HealthChecker healthChecker = new HealthChecker();
    private ZMQ.Context context;
    private Waitable waitable;

    public void setWaitable( Waitable waitable ) {
        this.waitable = waitable;
    }

    public ZmqServerRegisterClient( Set<String> clients, int ioThreads ) {
        if ( null == clients || 0 == clients.size() )
            throw new IllegalArgumentException( "clients list can't be null or empty!" );
        this.clients = clients;
        this.context = ZMQ.context( ioThreads );
    }

    @Override
    public boolean register( String bizType, String address ) {
        synchronized ( registered ) {
            List<String> biz = registered.get( address );
            if ( null == biz ) {
                biz = new ArrayList<>();
                registered.put( address, biz );
            }
            biz.add( bizType );
            return doOps( bizType, address, Ops.REGISTER );
        }
    }

    @Override
    public boolean unregister( String bizType, String address ) {
        synchronized ( registered ) {
            List<String> biz = registered.get( address );
            if ( null != biz ) {
                biz.remove( bizType );
            }
            return doOps( bizType, address, Ops.UNREGISTER );
        }
    }

    public void resetHeartbeatDura( long ms ) {
        healthChecker.checkingDuran.set( ms );
    }

    private boolean doOps( String bizType, String address, Ops ops ) {
        boolean allSuccess = true;
        for ( String remote : clients ) {
            allSuccess = allSuccess && doOps0( remote, bizType, address, ops );
        }
        return allSuccess;
    }

    private boolean doOps0( String remote, String bizType, String address, Ops ops ) {
        boolean success = true;
        ZMQ.Socket st = null;
        try {
            st = context.socket( ZMQ.REQ );
            st.connect( remote );
            byte[] req = ZmqRegisterReq.newBuilder()
                    .setMagicCode( MAGIC_CODE )
                    .setOps( ops )
                    .setAddress( address )
                    .setBizType( bizType ).build().toByteArray();
            byte[] reply = ProtoHelper.sendAndWaitRsp( st, req, 1000 );
            ZmqRegisterRsp rsp = ZmqRegisterRsp.parseFrom( reply );
            success = null != rsp && ( rsp.getMagicCode() == MAGIC_CODE );
        } catch ( Exception err ) {
            log.error( " error during connecting {} : {} ", remote, err );
            success = false;
        } finally {
            try {
                if ( null != st ) {
                    st.close();
                }
            } catch ( Exception err ) {
                log.error( "error when closing zmq socket: {} ", PlainUtil.getStackTrace( err ) );
            }
        }
        return success;
    }

    @Override
    public boolean isLaunchNow() {
        return true;
    }

    @Override
    public void run() {
        healthChecker.run();
    }

    class HealthChecker implements Runnable {

        AtomicLong checkingDuran = new AtomicLong( 1000 );

        @Override
        public void run() {
            if ( null != waitable ) waitable.waiting();
            while ( isKeepOn() ) {
                try {
                    pubOps( Ops.REGISTER );
                } catch ( Exception err ) {
                    log.error( "error in health checker: ", err );
                }
                try {
                    TimeUnit.MILLISECONDS.sleep( checkingDuran.get() );
                } catch ( InterruptedException ignored ) {
                }
            }
            try {
                pubOps( Ops.UNREGISTER );
            } catch ( Exception err ) {
                log.error( "error when unregister server: ", err );
            }
        }

        private void pubOps( Ops ops ) {
            for ( String cli : clients ) {
                synchronized ( registered ) {
                    for ( Map.Entry<String, List<String>> e : registered.entrySet() ) {
                        String address = e.getKey();
                        List<String> allBiz = e.getValue();
                        if ( null != allBiz ) {
                            for ( String biz : allBiz ) {
                                if ( !doOps0( cli, biz, address, ops ) ) {
                                    log.error( "Heart beat broken! remote: {} bizType: {} Ops: {} ", cli, biz, ops );
                                }
                            }
                        }
                    }
                }
            }
        }

    }


}
